package com.hzh.flink.core

import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo7Process {
  def main(args: Array[String]): Unit = {
    /**
     * 创建一个环境
     */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    /**
     * 读取socket的数据
     */
    val lines: DataStream[String] = env.socketTextStream("master", 8888)

    /**
     * process：一行一行处理数据，可以返回多行，相当于flatMap
     * process：可以用于代替map，filter，flatMap
     */
    val ProcessDS: DataStream[(String, Int)] = lines.process(new ProcessFunction[String, (String, Int)] {
//      /**
//       * 也可以重写open和close方法，可以将初始化的代码放在open中
//       * @param parameters
//       */
//      override def open(parameters: Configuration): Unit = {
//
//      }
//
//      override def close(): Unit = {
//
//
//      }

      /**
       * processElement: 将DS的数据一条一条传递给processElement
       *
       * @param value  ：一行数据
       * @param ctx      ：上下文对象,可以获取到flink的时间
       * @param out      ：用于将数据发送到下游
       */
      override def processElement(value: String,
                                  ctx: ProcessFunction[String, (String, Int)]#Context,
                                  out: Collector[(String, Int)]): Unit = {


        val clazz: String = value.split(",")(4)
        //将数据发送到下游
        out.collect((clazz, 1))

      }
    })

    ProcessDS.print()
    env.execute()





  }
}
